package com.movie.service.Impl;

import com.movie.service.SparkService;
import com.movie.utils.InputStreamReaderRunnable;
import org.apache.spark.launcher.SparkLauncher;
import org.springframework.stereotype.Service;

import java.util.HashMap;

@Service
public class SparkServiceImpl implements SparkService {


    public String submit(Long userId) {
        HashMap env = new HashMap();
        //hadoop、spark环境变量读取
        env.put("HADOOP_CONF_DIR", System.getenv().getOrDefault("HADOOP_CONF_DIR", "/data/soft/new/hadoop/etc/hadoop"));
        env.put("JAVA_HOME", System.getenv().getOrDefault("JAVA_HOME", "/data/soft/new/jdk1.8/jdk1.8.0_271"));
        //创建SparkLauncher启动器
        SparkLauncher handle = new SparkLauncher(env)
                .setSparkHome("/home/hadoop/spark/spark-2.1.1")
                .setAppResource("/home/hadoop/data/springspark/testOne.jar")
                .setMainClass("train.testOne")
                .setMaster("yarn")
                .setDeployMode("client")
                .setConf("spark.yarn.queue", "sino")
                .setConf("spark.app.id", "luncher-test")
                .setConf("spark.driver.memory", "1g")
                .setConf("spark.executor.memory", "1g")
                .setConf("spark.executor.instances", "2")
                .setConf("spark.executor.cores", "2")
                .setConf("spark.default.parallelism", "12")
                .setConf("spark.driver.allowMultipleContexts", "true")
                .setVerbose(true)
                .addAppArgs();

        try {
            //任务提交
            Process process = handle.launch();
            //日志跟踪子线程
            InputStreamReaderRunnable inputStreamReaderRunnable = new InputStreamReaderRunnable(process.getInputStream(), "input");
            Thread inputThread = new Thread(inputStreamReaderRunnable, "LogStreamReader input");
            inputThread.start();

            InputStreamReaderRunnable errorStreamReaderRunnable = new InputStreamReaderRunnable(process.getErrorStream(), "error");
            Thread errorThread = new Thread(errorStreamReaderRunnable, "LogStreamReader error");
            errorThread.start();

            System.out.println("Waiting for finish...");
            int exitCode = process.waitFor();
            System.out.println("Finished! Exit code:" + exitCode);
            return "status: " + exitCode;

        } catch (Exception e) {
            e.printStackTrace();
            return "status: " + 1;
        }

    }
}
